package com.itcast.spark.test

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.flume.{FlumeUtils, SparkFlumeEvent}
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies, OffsetRange}

/**
 * DESC:
 */
object kafka01 {
  def main(args: Array[String]): Unit = {
    //1-准备环境
    val conf: SparkConf = new SparkConf().setAppName("SparkStreamingTCPTopK").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    //这里就是指定配置项将数据按照5秒为周期进行处理
    val ssc = new StreamingContext(sc, Seconds(5))
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "node01:9092,node02:9092,node03:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "use_a_separate_group_id_for_each_stream",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    //2-读取数据
    val receive: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](Array("kafkatopic"), kafkaParams))
    val data: DStream[String] = receive.map(_.value())
    //获取状态
    receive.foreachRDD(rdd=>{
      if(rdd.count()>0){
        //可以打印topic中的内容
        rdd.foreach(f=>{
          println("topic info is:",f)
          println("topic value is:",f.value())
        })//end rdd
        //获取offset
        val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        //打印具体的额开始位置的offset到结束位置的offset
        for (o<-offsetRanges){
          println(s"topic:${o.topic} partiton:${o.partition} startoffset:${o.fromOffset} utiloffset:${o.untilOffset}")
        }
        //提交到offset到ck
        receive.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
      }//end if
    })
    //3-窗口的统计
    val result: DStream[(String, Int)] = data.flatMap(_.split(" ")).map(x => (x, 1)).reduceByKeyAndWindow((x: Int, y: Int) => x + y, Seconds(10), Seconds(5))
    //4-结果输出print
    result.print()
    //5-ssc.start
    ssc.start()
    //6-ssc.awaitTermination
    ssc.awaitTermination()
    //7-ssc.stop(true,true)
    ssc.stop(true, true)
  }
}
